package com.zhu.tool_test.starke.framework.redis.controller;

import com.zhu.tool_test.starke.util.lang.Strings;
import lombok.Setter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * @author  4everlynn
 * @version V1.0
 * @description 发送控制消息
 * @date 2019-04-16
 */
@Setter
public class EventController {
    private Integer rpcTimeout;

    @Autowired
    @Qualifier("starke-string-redis-tmpl")
    private RedisTemplate<String, Object> redisTemplate;

    /**
     * 发送没有返回值的消息
     *
     * @param queue   目标队列
     * @param command 命令
     */
    public void  sendCommand(String queue, Command command) {
        redisTemplate.opsForList()
                .leftPush(queue, command.build());
    }

    /**
     * 发送异步的、带返回值的消息
     *
     * @param queue           目标队列
     * @param command         命令
     * @param commandCallback 命令回调
     */
    public String asyncCommand(String queue, Command command, CommandCallback commandCallback) throws ExecutionException, InterruptedException {
        this.sendCommand(queue, command);

        // 实际上只有当命令带了reto的时候才会是异步的
        if (!Strings.isNullOrEmpty(command.returnToQueue())) {
            // 不考虑返回顺序，先得到的结果优先处理
            CompletableFuture<String> completableFuture
                    = CompletableFuture.supplyAsync(() -> (String) redisTemplate.opsForList()
                    // 默认10秒超时
                    // 阻塞线程,直到收到回复
                    // 超时事件作为参数传进
                    .rightPop(command.returnToQueue(), rpcTimeout, TimeUnit.SECONDS));
            return completableFuture.whenComplete((result, e) -> {
                if (null != commandCallback && null != result) {
                    try {
                        commandCallback.receiveData(result);
                    } catch (ExecutionException | InterruptedException e1) {
                        e1.printStackTrace();
                    }
                }
            }).get();
        }
        return null;
    }
}
